package com.emc.greenplum.pigsterix.compiler;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.commons.lang3.mutable.Mutable;

import org.apache.pig.impl.logicalLayer.FrontendException;
import org.apache.pig.newplan.logical.expression.ProjectExpression;
import org.apache.pig.newplan.logical.relational.LogicalPlan;

import com.emc.greenplum.pigsterix.compiler.metadata.PigsterixMetadataProvider;
import com.emc.greenplum.pigsterix.compiler.rewriter.PigsterixRewriteRuleset;
import com.emc.greenplum.pigsterix.compiler.rules.ConvertAlgebricks2MapReduceRule;

import edu.uci.ics.hyracks.algebricks.common.constraints.AlgebricksPartitionConstraint;
import edu.uci.ics.hyracks.algebricks.common.exceptions.AlgebricksException;
import edu.uci.ics.hyracks.algebricks.common.utils.Pair;
import edu.uci.ics.hyracks.algebricks.compiler.api.HeuristicCompilerFactoryBuilder;
import edu.uci.ics.hyracks.algebricks.compiler.api.ICompiler;
import edu.uci.ics.hyracks.algebricks.compiler.api.ICompilerFactory;
import edu.uci.ics.hyracks.algebricks.compiler.rewriter.rulecontrollers.SequentialFixpointRuleController;
import edu.uci.ics.hyracks.algebricks.compiler.rewriter.rulecontrollers.SequentialOnceRuleController;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.IHyracksJobBuilder;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalExpression;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalOperator;
import edu.uci.ics.hyracks.algebricks.core.algebra.base.ILogicalPlan;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IExpressionEvalSizeComputer;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IExpressionTypeComputer;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.ILogicalExpressionJobGen;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.INullableTypeComputer;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IPartialAggregationTypeComputer;
import edu.uci.ics.hyracks.algebricks.core.algebra.expressions.IVariableTypeEnvironment;
import edu.uci.ics.hyracks.algebricks.core.algebra.metadata.IMetadataProvider;
import edu.uci.ics.hyracks.algebricks.core.algebra.prettyprint.LogicalOperatorPrettyPrintVisitor;
import edu.uci.ics.hyracks.algebricks.core.algebra.prettyprint.PlanPrettyPrinter;
import edu.uci.ics.hyracks.algebricks.core.algebra.typing.ITypingContext;
import edu.uci.ics.hyracks.algebricks.core.config.AlgebricksConfig;
import edu.uci.ics.hyracks.algebricks.core.jobgen.impl.ConnectorPolicyAssignmentPolicy;
import edu.uci.ics.hyracks.algebricks.core.jobgen.impl.JobBuilder;
import edu.uci.ics.hyracks.algebricks.core.jobgen.impl.JobGenContext;
import edu.uci.ics.hyracks.algebricks.core.jobgen.impl.PlanCompiler;
import edu.uci.ics.hyracks.algebricks.core.rewriter.base.AbstractRuleController;
import edu.uci.ics.hyracks.algebricks.core.rewriter.base.IAlgebraicRewriteRule;
import edu.uci.ics.hyracks.algebricks.data.IBinaryBooleanInspector;
import edu.uci.ics.hyracks.algebricks.data.IBinaryComparatorFactoryProvider;
import edu.uci.ics.hyracks.algebricks.data.IBinaryHashFunctionFactoryProvider;
import edu.uci.ics.hyracks.algebricks.data.IBinaryIntegerInspector;
import edu.uci.ics.hyracks.algebricks.data.INormalizedKeyComputerFactoryProvider;
import edu.uci.ics.hyracks.algebricks.data.IPrinterFactoryProvider;
import edu.uci.ics.hyracks.algebricks.data.ISerializerDeserializerProvider;
import edu.uci.ics.hyracks.algebricks.data.ITypeTraitProvider;
import edu.uci.ics.hyracks.api.dataflow.value.INullWriterFactory;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.ITypeTraits;
import edu.uci.ics.hyracks.api.job.JobSpecification;
import edu.uci.ics.hyracks.api.topology.ClusterTopology;

public class PigsterixCompiler {
	private static final Logger LOGGER = Logger
			.getLogger(PigsterixCompiler.class.getName());

	private static List<Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>> buildDefaultLogicalRewrites() {
		List<Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>> defaultLogicalRewrites = new ArrayList<Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>>();
		SequentialFixpointRuleController seqCtrlNoDfs = new SequentialFixpointRuleController(
				false);
		SequentialFixpointRuleController seqCtrlFullDfs = new SequentialFixpointRuleController(
				true);
		SequentialOnceRuleController seqOnceCtrl = new SequentialOnceRuleController(
				true);
		defaultLogicalRewrites
				.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(
						seqOnceCtrl, PigsterixRewriteRuleset
								.buildTypeInferenceRuleCollection()));
		defaultLogicalRewrites
				.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(
						seqCtrlFullDfs, PigsterixRewriteRuleset
								.buildNormalizationRuleCollection()));
		defaultLogicalRewrites
				.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(
						seqCtrlNoDfs, PigsterixRewriteRuleset
								.buildCondPushDownRuleCollection()));
		defaultLogicalRewrites
				.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(
						seqCtrlNoDfs, PigsterixRewriteRuleset
								.buildJoinInferenceRuleCollection()));
		defaultLogicalRewrites
				.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(
						seqCtrlNoDfs, PigsterixRewriteRuleset
								.buildOpPushDownRuleCollection()));
		defaultLogicalRewrites
				.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(
						seqOnceCtrl, PigsterixRewriteRuleset
								.buildDataExchangeRuleCollection()));
		defaultLogicalRewrites
				.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(
						seqCtrlNoDfs, PigsterixRewriteRuleset
								.buildConsolidationRuleCollection()));
		return defaultLogicalRewrites;
	}

	private static List<Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>> buildDefaultPhysicalRewrites() {
		List<Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>> defaultPhysicalRewrites = new ArrayList<Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>>();
		SequentialOnceRuleController seqOnceCtrlAllLevels = new SequentialOnceRuleController(
				true);
		SequentialOnceRuleController seqOnceCtrlTopLevel = new SequentialOnceRuleController(
				false);
		defaultPhysicalRewrites
				.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(
						seqOnceCtrlAllLevels, PigsterixRewriteRuleset
								.buildPhysicalRewritesAllLevelsRuleCollection()));
		defaultPhysicalRewrites
				.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(
						seqOnceCtrlTopLevel, PigsterixRewriteRuleset
								.buildPhysicalRewritesTopLevelRuleCollection()));
		defaultPhysicalRewrites
				.add(new Pair<AbstractRuleController, List<IAlgebraicRewriteRule>>(
						seqOnceCtrlAllLevels, PigsterixRewriteRuleset
								.prepareForMapReduceJobGenRUleCollection()));
		return defaultPhysicalRewrites;
	}

	private final ICompilerFactory cFactory;

	private final PigsterixMetadataProvider metadataProvider;

	private int varCounter;

	private ILogicalOperator previousOp;

	public PigsterixCompiler() {
		HeuristicCompilerFactoryBuilder builder = new HeuristicCompilerFactoryBuilder();
		builder.setLogicalRewrites(buildDefaultLogicalRewrites());
		builder.setPhysicalRewrites(buildDefaultPhysicalRewrites());
		builder.setSerializerDeserializerProvider(new ISerializerDeserializerProvider() {
			@SuppressWarnings("unchecked")
			@Override
			public ISerializerDeserializer getSerializerDeserializer(Object type)
					throws AlgebricksException {
				return null;
			}
		});
		builder.setTypeTraitProvider(new ITypeTraitProvider() {
			public ITypeTraits getTypeTrait(Object type) {
				return null;
			}
		});
		// builder.setPrinterProvider(PigsterixPrinterFactoryProvider.INSTANCE);
		// builder.setExpressionRuntimeProvider(new
		// LogicalExpressionJobGenToExpressionRuntimeProviderAdapter(
		// new PigsterixExpressionJobGen()));
		builder.setExpressionTypeComputer(new IExpressionTypeComputer() {
			@Override
			public Object getType(ILogicalExpression expr,
					IMetadataProvider<?, ?> metadataProvider,
					IVariableTypeEnvironment env) throws AlgebricksException {
				return null;
			}
		});
		cFactory = builder.create();
		metadataProvider = new PigsterixMetadataProvider();
	}

	public JobSpecification compile(LogicalPlan pigPlan)
			throws AlgebricksException, FrontendException {
		Pig2AlgebricksVisitor translatorPig = new Pig2AlgebricksVisitor(pigPlan);

		// 2. Traverses the graph
		translatorPig.visit();

		ILogicalPlan plan = translatorPig.getAlgebricksPlan();

		if (LOGGER.isLoggable(Level.INFO)) {
			LOGGER.info("Translated Plan:");
			LOGGER.info(getPrettyPrintedPlan(plan));
		}
		ICompiler compiler = cFactory.createCompiler(plan, metadataProvider,
				varCounter);
		compiler.optimize();
		if (LOGGER.isLoggable(Level.INFO)) {
			LOGGER.info("Optimized Plan:");
			LOGGER.info(getPrettyPrintedPlan(plan));
		}
		
		LOGGER.info("Now building the super-nodes tree from the optimized plan:");
		
		//Iterate on each root
		List<Mutable<ILogicalOperator>> list = plan.getRoots();

		generateJobBuilder(plan, compiler);
		
		//??
		Object appContext = null;
		return compiler.createJob(appContext);
		
		
	}


	//Create a Jobbuilder object and submit it.
	//for each subtree, create a jobBuilder and wrap it into a MR. Then, connect between the Job builders and 
	private JobSpecification generateJobBuilder(ILogicalPlan plan, ICompiler compiler) {
		
        Object appContext = null;
		IMetadataProvider<?, ?> metadata = null;
		IBinaryHashFunctionFactoryProvider hashFunctionFactoryProvider = null;
		ISerializerDeserializerProvider serializerDeserializerProvider = null;
		IBinaryComparatorFactoryProvider comparatorFactoryProvider = null;
		IBinaryIntegerInspector binaryIntegerInspector = null;
		ITypeTraitProvider typeTraitProvider = null;
		IPrinterFactoryProvider printerProvider = null;
		IBinaryBooleanInspector binaryBooleanInspector = null;
		INullWriterFactory nullWriterFactory = null;
		ILogicalExpressionJobGen exprJobGen = null;
		INormalizedKeyComputerFactoryProvider normalizedKeyComputerFactoryProvider = null;
		IExpressionEvalSizeComputer expressionEvalSizeComputer = null;
		INullableTypeComputer nullableTypeComputer = null;
		ITypingContext oc = null;
		IExpressionTypeComputer expressionTypeComputer = null;
		IPartialAggregationTypeComputer partialAggregationTypeComputer = null;
		ClusterTopology clusterTopology = null;
		AlgebricksPartitionConstraint clusterLocations = null;
		int frameSize = 0;
		
		JobGenContext context = new JobGenContext(null, metadata, appContext,
                serializerDeserializerProvider, hashFunctionFactoryProvider, comparatorFactoryProvider,
                typeTraitProvider, binaryBooleanInspector, binaryIntegerInspector, printerProvider,
                nullWriterFactory, normalizedKeyComputerFactoryProvider, exprJobGen,
                expressionTypeComputer, nullableTypeComputer, oc, expressionEvalSizeComputer,
                partialAggregationTypeComputer, frameSize, clusterLocations, clusterTopology);
        
		PlanCompiler pc = new PlanCompiler(context);
        JobSpecification jobSpec = null;
		try {
			jobSpec = pc.compilePlan(plan, null);
		} catch (AlgebricksException e) {
			e.printStackTrace();
		}
		JobBuilder job = new JobBuilder(jobSpec, clusterLocations, clusterTopology);
		
		try {
			return compiler.createJob(job);
		} catch (AlgebricksException e) {
			e.printStackTrace();
		}
		return jobSpec;
	}

	private String getPrettyPrintedPlan(ILogicalPlan plan)
			throws AlgebricksException {
		StringBuilder buffer = new StringBuilder();
		PlanPrettyPrinter.printPlan(plan, buffer,
				new LogicalOperatorPrettyPrintVisitor(), 2);
		return buffer.toString();
	}

}
